package com.bytes32.preprocess;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.omg.CORBA.UNKNOWN;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.bytes32.classification.tree.dataset.ByteArrayWrapper;
import com.bytes32.classification.tree.dataset.DefaultFeatureProcessor;
import com.bytes32.classification.tree.dataset.Feature;
import com.bytes32.classification.tree.dataset.FeatureType;
import com.bytes32.classification.tree.dataset.InvalidFeatureValueException;
import com.bytes32.config.ParamsConfig;
import com.bytes32.parsing.Parser;
import com.bytes32.parsing.ParsingException;

/**
 * FIXME: everything works with strings we need a generic wrapper which can
 * encode/decode any kind of binary/non-binary format
 * 
 * @author murariuf
 * 
 */
public class FeatureCombiner implements Reducer<IntWritable, Feature, IntWritable, Feature> {

	/**
	 * Default Logger
	 */
	private static final Logger LOG = LoggerFactory.getLogger(FeatureCombiner.class);
	
	/*
	 * Each parser will run against every value
	 */
	@SuppressWarnings("rawtypes")
	private List<Parser> parsers;

 	/*
	 * In order to determine the type ORDERED/UNORDERED check out 100 items
	 */
	private int evaluate = 100;

	/*
	 * The first parser to reach 20 decides the type ORDERED or UNORDERED
	 */
	private int confidence = 20;
	
	@SuppressWarnings("rawtypes")
	@Override
	public void configure(JobConf conf) {
		/* get how confident we are with the evaluation */
		this.confidence = conf.getInt(ParamsConfig.PARSE_SUCCESS_COUNT.name(), 20);
		/* how many items do we look at */
		this.evaluate = conf.getInt(ParamsConfig.PARSE_BUFFER_SIZE.name(), 100);

		/* set the parsers */
		String[] parserNames = conf.getStrings(ParamsConfig.MAP_PARSERS.name());
		parsers = new ArrayList<Parser>();

		for (int i = 0; i < parserNames.length; i++) {
			String parser = parserNames[i];
			try {
				Class<?> clazz = Class.forName(parser);
				if (Parser.class.isAssignableFrom(clazz)) {
					Class<Parser> parserClass = (Class<Parser>) clazz;
					/* we have a parser */
					Constructor<Parser> construct = parserClass.getConstructor();
					Parser newParser = construct.newInstance();
					newParser.setId(i);
					parsers.add(newParser);
				}
			} catch (Exception e) {
				/* configure them right or remove them */
				throw new IllegalArgumentException(" Unable to create parsers ");
			}
		}

	}

	@Override
	public void close() throws IOException {

	}

	@Override
	public void reduce(IntWritable id, Iterator<Feature> values, OutputCollector<IntWritable, Feature> out, Reporter rep)
			throws IOException {
		/*
		 * attempt to understand what kind of feature we're dealing with you should
		 * determine the FeatureType ordered or un-ordered and the attribute type
		 * which can be long,double, binary or text
		 * 
		 * The assumption is that Numbers are ORDERED and String is UNORDERED
		 */

		/*
		 * count this.evaluate items, the first parser to get this.confidence votes
		 * wins
		 */
		FeatureBuilder builder = new FeatureBuilder(this.evaluate, this.confidence, parsers);

		/* at this point in the combiner every feature should be UNKNOWN with the DefaultFeatureProcessor */
		int i = 0;
		Feature result = null;
		while (values.hasNext()) {
			try {
				Feature value = values.next();
				/* we limit the types of features we can evaluate here */
				if (value.getFeatureType() != FeatureType.UNKNOWN)
					throw new IllegalStateException(" value is not unkonwn ");
				
				if (!(value.getProcessor() instanceof DefaultFeatureProcessor))
					throw new IllegalStateException(" invalid processor ");
				
				Iterator<Entry<ByteArrayWrapper, Long>> iter = ((DefaultFeatureProcessor)value.getProcessor()).iterator();
				
				String strVal = new String(iter.next().getKey().getWrapped(),"UTF-8");
				
				if (i < this.evaluate) {
					builder.evaluate(strVal);
					i++;
					if (i == this.evaluate) {
						result = builder.build();
					}
				}
				if (result == null){
					continue;
				} else {
					result.incrementCount(strVal, parsers);
				}
			} catch (ParsingException e) {
				LOG.error("Unable to parse value ",e);
			} catch (InvalidFeatureValueException e) {
				LOG.error("Unable to count value",e);
			}
		}
		result.compact();
		/* output the feature */
		out.collect(id, result);
	}
}
